package cn.t.util.casestudy.test.reactor;

import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Optional;
import java.util.Random;

public class ReactorApiCase {

    @Test
    public void createFluxTest() {
//        Flux.just("Hello", "World").subscribe(System.out::println);
//        Flux.fromArray(new Integer[] {1, 2, 3}).subscribe(System.out::println);
//        Flux.empty().subscribe(System.out::println);
//        Flux.range(1, 10).subscribe(System.out::println);
//        Flux.interval(Duration.of(1, ChronoUnit.SECONDS)).subscribe(System.out::println);
//        try {
//            Thread.sleep(10000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        Flux.create(sink -> {
            for(int i=0; i<5; i++) {
                sink.next(i);
            }
        }).subscribe(System.out::println);
    }

    @Test
    public void generateFluxTest() {
        Flux.generate(sink -> {
            sink.next("Hello");
            sink.complete();
        }).subscribe(System.out::println);

        final Random random = new Random();
        Flux.generate(ArrayList::new, (list, sink) ->{
            int value = random.nextInt(100);
            list.add(value);
            sink.next(value);
            if(list.size() == 3) {
                sink.complete();
            }
            return list;
        }).subscribe(System.out::println);
    }


    @Test
    public void createMonoTest() {
        Mono.fromSupplier(() -> "Hello").subscribe(System.out::println);
        Mono.justOrEmpty(Optional.of("Hello")).subscribe(System.out::println);
        Mono.create(sink -> sink.success("Hello")).subscribe(System.out::println);
    }

    @Test
    public void operateTest() {
//        Flux.range(1, 100).buffer(20).subscribe(System.out::println);
//        Flux.interval(Duration.ofMillis(100)).buffer(Duration.ofMillis(1000)).take(3).toStream().forEach(System.out::println);
//        Flux.range(1, 10).bufferUntil(i -> i % 2 == 0).subscribe(System.out::println);
//        Flux.range(1, 10).bufferWhile(i -> i<5 || i>6).subscribe(System.out::println);
    }

    @Test
    public void filterTest() {
        Flux.range(1, 10).filter(i -> i % 2 == 0).subscribe(System.out::println);
    }

    @Test
    public void windowTest() {
        Flux.range(1, 10).window(2).subscribe(flux-> {flux.subscribe(System.out::println);System.out.println("\r\n");});
    }

    @Test
    public void zipWithTest() {
//        Flux.just("a", "b")
//            .zipWith(Flux.just("c", "d"))
//            .subscribe(System.out::println);

        Flux.just("a", "b", "e")
            .zipWith(Flux.just("c", "d", "f"), (s1, s2) -> String.format("%s-%s", s1, s2))
            .subscribe(System.out::println);
    }

    @Test
    public void reduceTest() {
//        Flux.range(1, 100).reduce((x, y) -> x + y).subscribe(System.out::println);
        Flux.range(1, 100).reduceWith(() -> 100, (x, y) -> x + y).subscribe(System.out::println);
    }

    @Test
    public void mergeTest() {
//        Flux.merge(Flux.interval(Duration.ofMillis(100)).take(5), Flux.interval(Duration.ofMillis(50), Duration.ofMillis(100)).take(5)).toStream().forEach(System.out::println);
        Flux.mergeSequential(Flux.interval(Duration.ofMillis(100)).take(5), Flux.interval(Duration.ofMillis(50), Duration.ofMillis(100)).take(5)).toStream().forEach(System.out::println);
    }

    @Test
    public void flatMapTest() {
        Flux.just(5, 10)
            .flatMap(x -> Flux.interval(Duration.ofMillis(x * 10), Duration.ofMillis(500)).take(x))
            .toStream()
            .forEach(System.out::println);

//        Flux.just(5, 10)
//            .flatMapSequential(x -> Flux.interval(Duration.ofMillis(x * 10), Duration.ofMillis(500)).take(x))
//            .onBackpressureDrop()
//            .toStream()
//            .forEach(System.out::println);
    }

    @Test
    public void concatMapTest() {
        Flux.just(5, 10)
            .concatMap(x -> Flux.interval(Duration.ofMillis(x * 10), Duration.ofMillis(500)).take(x))
            .toStream()
            .forEach(System.out::println);
    }

    @Test
    public  void combineLatestTest() {
        Flux.combineLatest(Arrays::toString,
            Flux.interval(Duration.ofMillis(1000)).take(5),
            Flux.interval(Duration.ofMillis(100), Duration.ofMillis(200)).take(5),
            Flux.interval(Duration.ofMillis(200), Duration.ofMillis(200)).take(5)
        ).toStream().forEach(System.out::println);
    }

    @Test
    public void concatWithTest() {
        Flux.just(1, 2)
            .concatWith(Mono.error(new IllegalStateException()))
            .subscribe(System.out::println, System.err::println);
    }

    @Test
    public void onErrorReturnTest() {
        Flux.just(1, 2)
            .concatWith(Mono.error(new IllegalStateException()))
            .onErrorReturn(0)
            .subscribe(System.out::println);
    }

    @Test
    public void switchOnErrorTest() {
        Flux.just(1, 2)
            .concatWith(Mono.error(new IllegalStateException()))
            .concatWith(Mono.just(11))
            .onErrorResume(e -> e instanceof IllegalStateException, e -> Mono.just(0))
            .subscribe(System.out::println);
    }


}
